package cn.monkey.transport.mq.rabbitmq;

import cn.monkey.transport.mq.rabbitmq.codec.Codec;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.rabbitmq.ConsumeOptions;
import reactor.rabbitmq.Receiver;

import java.util.function.Predicate;

public class SimpleRabbitMqInbound implements RabbitMqInbound {
    protected final Receiver receiver;
    protected final Codec codec;
    protected final String queue;
    protected final ConsumeOptions consumeOptions;

    public SimpleRabbitMqInbound(Receiver receiver, Codec codec, String queue, ConsumeOptions consumeOptions) {
        this.receiver = receiver;
        this.queue = queue;
        this.consumeOptions = consumeOptions;
        this.codec = codec;
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz, Predicate<T> p) {
        return this.receiver.consumeAutoAck(this.queue, this.consumeOptions)
                .map(delivery -> {
                    byte[] body = delivery.getBody();
                    return this.codec.decode(body);
                })
                .filter(clazz::isInstance)
                .map(clazz::cast)
                .filter(p);
    }

    @Override
    public Mono<Void> close() {
        this.receiver.close();
        return Mono.empty();
    }


    @Override
    public String topic() {
        return this.queue;
    }

    @Override
    public Codec codec() {
        return this.codec;
    }

    @Override
    public ConsumeOptions consumeOptions() {
        return this.consumeOptions;
    }
}
